package com.shujia.flink.state

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.api.common.state.{ValueState, ValueStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

object demo7FraudState {
  def main(args: Array[String]): Unit = {
    /**
     * 对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。
     * 001,20.1
     * 001,30
     * 001,0.07
     * 001,100
     * 001,800
     * 001,20
     * 001,0.1
     * 001,700
     * 001,20
     * 001,0.1
     * 002,700
     * 001,800
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val linesDS: DataStream[String] = env.socketTextStream("master", 8888)

    //解析数据
    val kvDS: DataStream[(String, Double)] = linesDS.map(line => {
      val split: Array[String] = line.split(",")
      val id: String = split(0)
      val price: Double = split(1).toDouble
      (id, price)
    })

    //安装账号分组
    val keyByDS: KeyedStream[(String, Double), String] = kvDS.keyBy(_._1)

    val fraudDS: DataStream[Fraud] = keyByDS
      .process(new KeyedProcessFunction[String, (String, Double), Fraud] {
        //用于保存标记的状态
        var flagState: ValueState[Boolean] = _
        var minPriceState: ValueState[Double] = _

        override def open(parameters: Configuration): Unit = {
          val context: RuntimeContext = getRuntimeContext

          //拆功能键状态描述对象，默认值为False
          val flagDesc = new ValueStateDescriptor[Boolean]("flag", classOf[Boolean], false)
          flagState = context.getState(flagDesc)

          //创建一个用于保存小于1每一金额的状态
          val minPriceDesc = new ValueStateDescriptor[Double]("minPrice", classOf[Double])
          minPriceState = context.getState(minPriceDesc)
        }

        override def processElement(value: (String, Double),
                                    ctx: KeyedProcessFunction[String, (String, Double), Fraud]#Context,
                                    out: Collector[Fraud]): Unit = {
          val (id: String, price: Double) = value
          //对于一个账户，如果出现小于 $1 美元的交易后紧跟着一个大于 $500 的交易，就输出一个报警信息。

          val flag: Boolean = flagState.value()
          if (flag) {
            if (price > 500) {
              //从状态中获取小于1美元的金额
              val minPrice: Double = minPriceState.value()
              //获取当前的时间
              val time: Long = System.currentTimeMillis()
              //发出告警信息
              out.collect(Fraud(id, time, minPrice, price))
            }
            //重置flag状态
            flagState.update(false)
          }
          if (price < 1) {
            //如果小于1美元出现了，将flag改成true
            flagState.update(true)
            //将小于1美元的金额保存在状态中
            minPriceState.update(price)

            //注册定时器，在一段时间后触发onTimer的执行
            val time: Long = ctx.timerService().currentProcessingTime()
            ctx.timerService().registerProcessingTimeTimer(time + 5000)
          }
        }

        override def onTimer(timestamp: Long, ctx: KeyedProcessFunction[String, (String, Double), Fraud]#OnTimerContext, out: Collector[Fraud]): Unit = {
          //将flag改成false
          flagState.update(false)
          //清除状态的数据
          minPriceState.clear()
        }
      })

    fraudDS.print()

    env.execute()
  }

  case class Fraud(id: String, time: Long, minPrice: Double, maxPrice: Double)


}
